Kafka 生产者和消费者:如何编写 Kafka 生产者和消费者

您所在的位置:网站首页 多个生产者 一个消费者怎么处理 Kafka 生产者和消费者:如何编写 Kafka 生产者和消费者

Kafka 生产者和消费者:如何编写 Kafka 生产者和消费者

2023-06-04 07:02| 来源: 网络整理| 查看: 265

Kafka 是一个高性能、分布式、基于发布/订阅模式的消息队列系统,它支持可靠地处理大量的数据流。在 Kafka 中,生产者将消息发布到主题(Topic),而消费者从主题中订阅消息。本文将介绍如何编写 Kafka 生产者和消费者。

Kafka 生产者

Kafka 生产者是将消息发布到主题的客户端。下面是一个简单的 Kafka 生产者示例:

import org.apache.kafka.clients.producer.*; import java.util.Properties; public class KafkaProducerExample { public static void main(String[] args) { String topicName = "test-topic"; String message = "Hello, Kafka!"; Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer producer = new KafkaProducer(props); ProducerRecord record = new ProducerRecord(topicName, message); producer.send(record); producer.close(); } }

上述代码中,我们首先定义了一个主题名和消息内容,然后创建了一个 Kafka 生产者。在创建 Kafka 生产者时,我们需要指定 Kafka 服务器的地址和端口,以及消息的 key 和 value 的序列化器。接着,我们创建了一个 ProducerRecord 对象,将主题名和消息内容传递给它,然后通过 Kafka 生产者的 send() 方法将消息发送到主题中。最后,我们需要关闭 Kafka 生产者。

Kafka 消费者

Kafka 消费者是从主题中订阅消息的客户端。下面是一个简单的 Kafka 消费者示例:

import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import java.util.Arrays; import java.util.Properties; public class KafkaConsumerExample { public static void main(String[] args) { String topicName = "test-topic"; Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer consumer = new KafkaConsumer(props); consumer.subscribe(Arrays.asList(topicName)); while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } } } }

上述代码中,我们首先定义了一个主题名,然后创建了一个 Kafka 消费者。在创建 Kafka 消费者时,我们需要指定 Kafka 服务器的地址和端口,以及消费者所属的消费组的名称,以及消息的 key 和 value 的反序列化器。接着,我们通过 Kafka 消费者的 subscribe() 方法订阅了主题。最后,我们通过一个无限循环不断地从主题中获取消息,然后处理这些消息。

总结

本文介绍了如何编写 Kafka 生产者和消费者。在实际的应用中,需要根据具体的需求来配置 Kafka 生产者和消费者。同时,还需要考虑到 Kafka 的高可用性和性能问题,以保证 Kafka 的稳定性和可靠性。



【本文地址】


今日新闻


推荐新闻


CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3